package cn._51doit.day09;

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.*;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.OutputTag;

/**
 * @create: 2021-10-26 22:56
 * @author: 今晚打脑斧先森
 * @program: GetWindowData
 * @Description:
 *  使用测流输出获取窗口迟到的数据
 *  一般processingTiem的数据不会有迟到的数据,除非极端情况,网络延迟对吧
 **/
public class GetWindowData {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //使用老的API
        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);

        //1000,flink,1
        DataStreamSource<String> lines = env.socketTextStream("doit01", 8888);
        SingleOutputStreamOperator<Tuple3<Long, String, Integer>> tpStream = lines.map(new MapFunction<String, Tuple3<Long, String, Integer>>() {
            @Override
            public Tuple3<Long, String, Integer> map(String value) throws Exception {
                String[] split = value.split(",");
                return Tuple3.of(Long.parseLong(split[0]), split[1], Integer.parseInt(split[2]));
            }
        }).setParallelism(1);

        //获取waterMark
        //窗口延迟时间0秒
        SingleOutputStreamOperator<Tuple3<Long, String, Integer>> wordAndCountWithMaterMark  = tpStream.assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor<Tuple3<Long, String, Integer>>(Time.seconds(0)) {
            @Override
            public long extractTimestamp(Tuple3<Long, String, Integer> element) {
                return element.f0;
            }
        });

        //先调用keyby
        KeyedStream<Tuple3<Long, String, Integer>, String> keyedStream = wordAndCountWithMaterMark.keyBy(t -> t.f1);

        //默认情况下,迟到的数据会被丢弃
        //所以给迟到的数据打上标签,因此,就能记录下来了
        OutputTag<Tuple3<Long, String, Integer>> lateDateTag = new OutputTag<Tuple3<Long, String, Integer>>("迟到数据"){};

        //划分窗口,打上标签
        WindowedStream<Tuple3<Long, String, Integer>, String, TimeWindow> windowedStream = keyedStream
                .window(TumblingEventTimeWindows.of(Time.seconds(10)))
                .sideOutputLateData(lateDateTag); //将窗口迟到的数据打上标签
        SingleOutputStreamOperator<Tuple3<Long, String, Integer>> res = windowedStream.sum(2);

        //获取迟到的数据，获取打标签的数据(主流中的按照是否打标签，分为2种，一种是未打标签的，一种是打标签的)
        DataStream<Tuple3<Long, String, Integer>> dataStream = res.getSideOutput(lateDateTag);
        dataStream.print("迟到数据: ");

        //正常数据
        res.print("正常数据: ");
        env.execute();
    }
}
